package com.bigdata.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1_MakeRDD {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("makeRdd")

    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    // sc.parallelize(集合，numSlices)
    val numRDD: RDD[Int] = sc.parallelize[Int](List[Int](1, 2, 3, 4, 5))

    // RDD 编写的相关代码，最终会多线程的分布式处理，spark框架来处理的
    // 在编写代码的时候，可以不用考虑分布式，就按照代理的逻辑编写就可以了
    // numRDD.foreach(println)

    //    sc.makeRDD(集合，numSlices)
    val num2RDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 1, 2), 2)

    println(s"分区数：${num2RDD.getNumPartitions}")

    val kvRDD: RDD[(Int, String)] = num2RDD.map((_, "abc"))

    // reduceByKey作为在k，v格式的RDD上的，分区数和上游的RDD保持一致
    val result: RDD[(Int, String)] = kvRDD.reduceByKey((v1: String, v2: String) => {
      v1 + v2
    })

    println(s"result分区数：${num2RDD.getNumPartitions}")

    // action算子才能触发job的执行
    result.foreach(println)



  }

}
